/**
 * 
 */
package jobs.step1;

import java.io.IOException;
import java.util.Date;
import java.util.Formatter;

import jobs.poc.sumMapper;
import jobs.poc.sumReducer;
import jobs.poc.writables.WordPairDecade;
import jobs.poc.writables.WordPairDecadeTimes;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

/**
 * @author zvikadori
 *
 */
public class Step1Driver {
	
	public static boolean runStep1() throws Exception{
		Configuration conf = new Configuration();	
		
		Job job = null;
		job = new Job(conf);
		
	
    
	    // Process custom command-line options
		Formatter formatter = new Formatter();
        String outpath = "Out"
                + formatter.format("%1$tm%1$td%1$tH%1$tM%1$tS", new Date());
	    Path in = new Path("In");
	    Path out = new Path(outpath);
	    
	    // Specify various job-specific parameters     
	    job.setJobName("sumMR");
	    SequenceFileInputFormat.addInputPath(job, in);
	    job.setInputFormatClass(TextInputFormat.class);

	    TextOutputFormat.setOutputPath(job, out);
	    job.setOutputFormatClass(TextOutputFormat.class);
	    
	    job.setMapperClass(Step1Mapper.class);
	    job.setMapOutputKeyClass(Text.class);
	    job.setMapOutputValueClass(LongWritable.class);
	    	    
	    job.setReducerClass(Step1Reducer.class);
	    job.setOutputKeyClass(Text.class);
	    job.setOutputValueClass(Text.class);
	
	    // Submit the job, then poll for progress until the job is complete
	    return job.waitForCompletion(true);
	}

}
